package org.agile4j.rpc.registry.zookeeper;

import org.I0Itec.zkclient.ZkClient;
import org.agile4j.rpc.registry.ServiceDiscovery;
import org.agile4j.rpc.registry.zookeeper.constant.Constant;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

/**
 * 使用 ZooKeeper 实现服务发现功能
 *
 * @author Administrator
 */
public class ZooKeeperServiceDiscovery implements ServiceDiscovery {

	private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperServiceDiscovery.class) ;

	private String zkAddress;

	public ZooKeeperServiceDiscovery(String zkAddress) {
		this.zkAddress = zkAddress;
	}

	@Override
	public String discover(String serviceName) {
		// 创建 ZooKeeper 客户端
		ZkClient zkClient = new ZkClient(zkAddress, Constant.ZK_SESSION_TIMEOUT, Constant.ZK_CONNECTION_TIMEOUT) ;
		LOGGER.debug("connect zookeeper");
		try {
			// 获取 service 节点
			String servicePath = Constant.ZK_REGISTRY_PATH + "/" + serviceName ;
			if (!zkClient.exists(servicePath)){
				throw new RuntimeException(String.format("can not find any service node on path: %s", servicePath)) ;
			}
			List<String> addressList = zkClient.getChildren(servicePath) ;
			if (CollectionUtils.isEmpty(addressList)) {
				throw new RuntimeException(String.format("can not find any address node on path: %s", servicePath)) ;
			}
			// 获取 address 节点
			String address ;
			int size = addressList.size() ;
			if (size == 1) {
				// 若只有一个地址，则获取该地址
				address = addressList.get(0) ;
				LOGGER.debug("get only address node: {}", address);
			} else {
				// 使用线程局部变量。每个线程希望以不同的生成器生成随机数，但它们是来自相同类的管理
				address = addressList.get(ThreadLocalRandom.current().nextInt(size)) ;
				LOGGER.debug("get random address node: {}", address);
			}
			// 获取 address 节点的值
			String addressPath = servicePath + "/" + address ;
			return zkClient.readData(addressPath) ;
		} finally {
			zkClient.close();
		}
	}

}
